package org.dding.msc.service.impl;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.management.ServiceNotFoundException;

import org.dding.core.log.ILogger;
import org.dding.core.log.LoggerFactory;
import org.dding.msc.exception.CircularDependencyException;
import org.dding.msc.exception.DuplicateServiceException;
import org.dding.msc.exception.ServiceLifecycException;
import org.dding.msc.service.DependencyInfo;
import org.dding.msc.service.Service;
import org.dding.msc.service.ServiceContainer;
import org.dding.msc.service.ServiceController;
import org.dding.msc.service.ServiceController.Mode;
import org.dding.msc.service.ServiceController.State;
import org.dding.msc.service.ServiceInfoWarehouse;
import org.dding.msc.service.ServiceInfoWarehouse.ServiceInfo;
import org.dding.msc.service.ServiceName;
import org.dding.msc.service.ServiceRegistration;
import org.dding.msc.service.ServiceRegistry;
import org.dding.msc.service.ServiceState;
import org.dding.msc.service.ServiceStateManager;
import org.dding.msc.service.listener.MultipleRemoveListener;
import org.dding.msc.value.UnlockedReadHashMap;
import org.jboss.modules.ref.Reaper;
import org.jboss.modules.ref.Reference;
import org.jboss.modules.ref.WeakReference;

public final class ServiceContainerImpl extends ServiceTargetImpl implements
		ServiceContainer {

	private static final AtomicInteger SERIAL = new AtomicInteger(1);

	private ILogger logger = LoggerFactory.getLogger(getClass());
	static final String PROFILE_OUTPUT;

	static {
		PROFILE_OUTPUT = AccessController
				.doPrivileged(new PrivilegedAction<String>() {
					public String run() {
						return System.getProperty("jboss.msc.profile.output");
					}
				});
	}

	private final ConcurrentMap<ServiceName, ServiceRegistration> registry = new UnlockedReadHashMap<ServiceName, ServiceRegistration>(
			512);

	private final long start = System.nanoTime();
	private long shutdownInitiated;

	private final List<TerminateListener> terminateListeners = new ArrayList<TerminateListener>(
			1);

	private static final class ShutdownHookHolder {
		private static final Set<Reference<ServiceContainerImpl, Void>> containers;
		private static boolean down = false;

		static {
			containers = new HashSet<Reference<ServiceContainerImpl, Void>>();
			AccessController.doPrivileged(new PrivilegedAction<Void>() {
				public Void run() {
					final Thread hook = new Thread(new Runnable() {
						public void run() {
							// shut down all services in all containers.
							final Set<Reference<ServiceContainerImpl, Void>> set = containers;
							final LatchListener listener;
							synchronized (set) {
								down = true;
								listener = new LatchListener(set.size());
								for (Reference<ServiceContainerImpl, Void> containerRef : set) {
									final ServiceContainer container = containerRef
											.get();
									if (container == null) {
										listener.countDown();
										continue;
									}
									container.addTerminateListener(listener);
									container.shutdown();
								}
								set.clear();
							}
							// wait for all services to finish.
							for (;;)
								try {
									listener.await();
									break;
								} catch (InterruptedException e) {
								}
						}
					}, "MSC Shutdown Thread");
					hook.setDaemon(false);
					Runtime.getRuntime().addShutdownHook(hook);
					return null;
				}
			});
		}

		private ShutdownHookHolder() {
		}
	}

	private final Writer profileOutput;

	private TerminateListener.Info terminateInfo = null;

	private volatile boolean down = false;

	private final ContainerExecutor executor;

	private final String name;

	public ServiceContainerImpl(String name, int coreSize, long timeOut,
			TimeUnit timeOutUnit) {
		super(null);
		final int serialNo = SERIAL.getAndIncrement();
		if (name == null) {
			name = String.format("anonymous-%d", Integer.valueOf(serialNo));
		}
		this.name = name;
		executor = new ContainerExecutor(coreSize, coreSize, timeOut,
				timeOutUnit);
		final Set<Reference<ServiceContainerImpl, Void>> set = ShutdownHookHolder.containers;
		Writer profileOutput = null;
		if (PROFILE_OUTPUT != null) {
			try {
				profileOutput = new OutputStreamWriter(new FileOutputStream(
						PROFILE_OUTPUT));
			} catch (FileNotFoundException e) {
				// ignore
			}
		}
		this.profileOutput = profileOutput;
		synchronized (set) {
			// if the shutdown hook was triggered, then no services can ever
			// come up in any new containers.
			if (ShutdownHookHolder.down) {
				down = true;
			}
			// noinspection ThisEscapedInObjectConstruction
			else {
				// noinspection ThisEscapedInObjectConstruction
				set.add(new WeakReference<ServiceContainerImpl, Void>(this,
						null, new Reaper<ServiceContainerImpl, Void>() {
							public void reap(final Reference<ServiceContainerImpl, Void> reference) {
								ShutdownHookHolder.containers.remove(reference);
							}
						}));
			}
		}
		// if (objectName != null) {
		// addTerminateListener(new TerminateListener() {
		// public void handleTermination(final Info info) {
		// try {
		// mBeanServer.unregisterMBean(objectName);
		// } catch (Exception ignored) {
		// }
		// }
		// });
		// }
	}

	public String getName() {
		return name;
	}

	public Writer getProfileOutput() {
		return profileOutput;
	}

	public long getStart() {
		return start;
	}

	@Override
	public synchronized void addTerminateListener(TerminateListener listener) {
		if (terminateInfo != null) { // if shutdown is already performed
			listener.handleTermination(terminateInfo); // invoke
														// handleTermination
														// immediately
		} else {
			terminateListeners.add(listener);
		}
	}

	@Override
	public void awaitTermination() throws InterruptedException {
		final LatchListener listener = new LatchListener(1);
		addTerminateListener(listener);
		listener.await();
	}

	@Override
	public void awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
		final LatchListener listener = new LatchListener(1);
		addTerminateListener(listener);
		listener.await(timeout, unit);
	}

	@Override
	public ServiceRegistry getServiceRegistry() {
		return this;
	}

	public boolean isShutdown() {
		return down;
	}

	public void shutdown() {
		final MultipleRemoveListener<Runnable> shutdownListener;
		synchronized (this) {
			if (down) {
				return;
			}
			down = true;
			shutdownInitiated = System.nanoTime();
		}
		shutdownListener = MultipleRemoveListener.create(new Runnable() {
			public void run() {
				executor.shutdown();
			}
		});
		final HashSet<ServiceController<?>> done = new HashSet<ServiceController<?>>();
		for (ServiceRegistration registration : registry.values()) {
			ServiceController<?> serviceInstance = registration.getInstance();
			if (serviceInstance != null
					&& serviceInstance.getState() != State.DOWN
					&& done.add(serviceInstance)) {
				try {
					serviceInstance.addListener(shutdownListener);
				} catch (IllegalArgumentException e) {
					continue;
				}
				serviceInstance.setMode(Mode.REMOVE);
			}
		}
		shutdownListener.done();
	}

	public boolean isShutdownComplete() {
		synchronized (this) {
			return terminateInfo != null;
		}
	}

	protected void finalize() throws Throwable {
		shutdown();
	}

	private synchronized void shutdownComplete(long started) {
		terminateInfo = new TerminateListener.Info(started, System.nanoTime());
		for (TerminateListener terminateListener : terminateListeners) {
			try {
				terminateListener.handleTermination(terminateInfo);
			} catch (Throwable t) {
				// ignore
			}
		}
	}

	static final class LatchListener extends CountDownLatch implements
			TerminateListener {

		LatchListener(int count) {
			super(count);
		}

		@Override
		public void handleTermination(Info info) {
			countDown();
		}
	}

	public Executor getExecutor() {
		return executor;
	}

	/**
	 * Atomically get or create a registration.
	 * 
	 * @param name
	 *            the service name
	 * @return the registration
	 */
	private ServiceRegistration getOrCreateRegistration(final ServiceName name) {
		final ConcurrentMap<ServiceName, ServiceRegistration> registry = this.registry;
		ServiceRegistration registration;
		registration = registry.get(name);
		if (registration == null) {
			registration = new ServiceRegistrationImpl(this, name);
			ServiceRegistration existing = registry.putIfAbsent(name,
					registration);
			if (existing != null) {
				return existing;
			} else {
				return registration;
			}
		} else {
			return registration;
		}
	}

	@Override
	public ServiceController<?> getRequiredService(final ServiceName serviceName) throws ServiceNotFoundException {
		final ServiceController<?> controller = getService(serviceName);
		if (controller == null) {
			throw new ServiceNotFoundException("Service " + serviceName
					+ " not found");
		}
		return controller;
	}

	@Override
	public ServiceController<?> getService(final ServiceName serviceName) {
		final ServiceRegistration registration = registry.get(serviceName);
		return registration == null ? null : registration.getInstance();
	}

	@Override
	public List<ServiceName> getServiceNames() {
		final List<ServiceName> result = new ArrayList<ServiceName>(
				registry.size());
		for (Map.Entry<ServiceName, ServiceRegistration> registryEntry : registry
				.entrySet()) {
			if (registryEntry.getValue().getInstance() != null) {
				result.add(registryEntry.getKey());
			}
		}
		return result;
	}

	void apply(ServiceBuilderImpl<?> builder) {
		// Apply listeners from the target, first
		super.apply(builder);
	}

	@Override
	<T> ServiceController<T> install(final ServiceBuilderImpl<T> serviceBuilder) throws DuplicateServiceException {
		if (down) {
			throw new IllegalStateException("Container is down");
		}
		apply(serviceBuilder);

		// Get names & aliases
		final ServiceName name = serviceBuilder.getName();
		final ServiceName[] aliases = serviceBuilder.getAliases();
		final int aliasCount = aliases.length;

		// Create registrations
		final ServiceRegistration primaryRegistration = getOrCreateRegistration(name);
		final ServiceRegistration[] aliasRegistrations = new ServiceRegistration[aliasCount];

		for (int i = 0; i < aliasCount; i++) {
			aliasRegistrations[i] = getOrCreateRegistration(aliases[i]);
		}

		// Next create the actual controller
		final ServiceControllerImpl<T> instance = new ServiceControllerImpl<T>(
				serviceBuilder.getServiceValue(), primaryRegistration,
				aliasRegistrations, serviceBuilder.getListeners());
		boolean ok = false;
		try {
			instance.startInstallation();
			// detect circularity before committing
			detectCircularity(instance);
			instance.commitInstallation(serviceBuilder.getInitialMode());
			afterCommit();
			ok = true;
			return instance;
		} finally {
			if (!ok) {
				instance.rollbackInstallation();
			}
		}
	}

	private void afterCommit() {
		ServiceStateManager manager = ServiceStateManager.getInstance();
		for (Entry<ServiceName, ServiceState> entry : manager
				.getServiceStates().entrySet()) {
			ServiceController<?> controller = getService(entry.getKey());
			if (controller != null)
				try {
					controller.setState(State.STARTING);
				} catch (ServiceLifecycException e) {
					logger.warn("start service "
							+ entry.getKey().getSimpleName() + " failure.", e);
				}
		}
	}

	@Override
	public boolean stopService(String serviceName) {
		ServiceController<?> controller = getService(ServiceName
				.of(serviceName));
		if (controller == null) {
			logger.warn("service " + serviceName + " dose not exist.");
			return false;
		}
		logger.debug("stop service " + serviceName);
		try {
			controller.setState(State.STOPPING);
			registry.remove(ServiceName.of(serviceName));
		} catch (ServiceLifecycException e) {
			logger.warn("stop service " + serviceName + " failure.", e);
			return false;
		}
		return true;
	}

	@Override
	public boolean startService(String serviceName) {
		ServiceController<?> controller = getService(ServiceName
				.of(serviceName));
		if (controller != null) {
			try {
				controller.setState(State.STARTING);
			} catch (ServiceLifecycException e) {
				logger.warn("start service " + serviceName + " failure.", e);
				return false;
			}
			return true;
		}

		logger.debug("start service " + serviceName);

		logger.info("service " + serviceName
				+ " dose not exist,will be reinstall.");
		ServiceInfo serviceInfo = ServiceInfoWarehouse.getInstance().get(
				serviceName);
		Class<? extends Service> serviceClazz = serviceInfo == null ? null
				: serviceInfo.serviceClass;
		if (serviceClazz == null) {
			logger.warn("start service " + serviceName
					+ " failure,can't find it");
			return false;
		}
		try {
			Service<?> service = serviceClazz.getConstructor().newInstance();
			addService(ServiceName.of(serviceName), service).install();
		} catch (Throwable e) {
			logger.warn("start service " + serviceName + " failure.", e);
			return false;
		}
		return true;
	}

	@Override
	public boolean removeService(String serviceName) {
		logger.debug("remove service " + serviceName);
		// ServiceController<?> controller = getService(ServiceName
		// .of(serviceName));
		//
		// if (controller == null) {
		// logger.warn("service " + serviceName + " dose not exist.");
		// return false;
		// }
		// try {
		// controller.setState(State.REMOVE);
		// registry.remove(ServiceName.of(serviceName));
		// } catch (ServiceLifecycException e) {
		// logger.warn("stop service " + serviceName + " failure.", e);
		// return false;
		// }
		return true;
	}

	@Override
	public boolean waitService(String serviceName) {
		ServiceController<?> controller = getService(ServiceName
				.of(serviceName));
		if (controller == null) {
			logger.warn("service " + serviceName + " dose not exist.");
			return false;
		}
		State targetState = State.WAITING;
		State state = controller.getState();
		if (state == State.WAITING) {
			targetState = State.UP;
		}

		if (targetState == State.WAITING)
			logger.debug("wait service " + serviceName);
		else
			logger.debug("enable service " + serviceName);

		try {
			controller.setState(targetState);
		} catch (ServiceLifecycException e) {
			logger.warn("wait service " + serviceName + " failure.", e);
			return false;
		}
		return true;
	}

	/**
	 * Detects if installation of {@code instance} results in dependency cycles.
	 * 
	 * @param instance
	 *            the service being installed
	 * @throws CircularDependencyException
	 *             if a dependency cycle involving {@code instance} is detected
	 */
	private <T> void detectCircularity(ServiceController<?> instance) throws CircularDependencyException {

		ServiceStateManager manager = ServiceStateManager.getInstance();
		Set<ServiceName> keys = manager.getServiceStates().keySet();
		for (Entry<ServiceName, ServiceState> entry : manager
				.getServiceStates().entrySet()) {

			ServiceName key = entry.getKey();
			ServiceState value = entry.getValue();

			for (DependencyInfo depInfo : value.getIDepends()) {
				if (!keys.contains(depInfo.getService()))
					continue;
				ServiceState targetService = manager.getServiceState(depInfo
						.getService());
				DependencyInfo dependencyInfo = targetService.getIDepend(key);
				if (dependencyInfo == null) {
					continue;
				} else {
					logger.warn("Between service[" + key.getSimpleName()
							+ "] and service["
							+ depInfo.getService().getSimpleName()
							+ "] exist depend cycly");
				}
			}
		}

	}

	private static final AtomicInteger executorSeq = new AtomicInteger(1);
	private static final Thread.UncaughtExceptionHandler HANDLER = new Thread.UncaughtExceptionHandler() {
		public void uncaughtException(final Thread t, final Throwable e) {
			// ServiceLogger.ROOT.uncaughtException(e, t);
		}
	};
	private static final ThreadPoolExecutor.CallerRunsPolicy POLICY = new ThreadPoolExecutor.CallerRunsPolicy();

	static class ServiceThread extends Thread {
		private final ServiceContainer container;

		ServiceThread(final Runnable runnable, final ServiceContainer container) {
			super(runnable);
			this.container = container;
		}

		ServiceContainer getContainer() {
			return container;
		}
	}

	public final class ContainerExecutor extends ThreadPoolExecutor {

		ContainerExecutor(final int corePoolSize, final int maximumPoolSize,
				final long keepAliveTime, final TimeUnit unit) {
			super(corePoolSize, maximumPoolSize, keepAliveTime, unit,
					new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
						private final int id = executorSeq.getAndIncrement();
						private final AtomicInteger threadSeq = new AtomicInteger(
								1);

						public Thread newThread(final Runnable r) {
							Thread thread = new ServiceThread(r,
									ServiceContainerImpl.this);
							thread.setName(String.format(
									"MSC service thread %d-%d", Integer
											.valueOf(id), Integer
											.valueOf(threadSeq
													.getAndIncrement())));
							thread.setUncaughtExceptionHandler(HANDLER);
							return thread;
						}
					}, POLICY);
		}

		protected void afterExecute(final Runnable r, final Throwable t) {
			super.afterExecute(r, t);
			if (t != null) {
				HANDLER.uncaughtException(Thread.currentThread(), t);
			}
		}

		protected void terminated() {
			shutdownComplete(shutdownInitiated);
		}
	}
}
